package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.OverallIdCreate;
import com.hmdp.utils.UserHolder;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private ISeckillVoucherService iSeckillVoucherService;
    @Resource
    private OverallIdCreate overallIdCreate;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    private final static DefaultRedisScript<Long>asyncMiaoShaScript;
    private final ExecutorService miaoshaExecutor= Executors.newSingleThreadExecutor();
    private BlockingQueue<VoucherOrder>blockingQueue=new ArrayBlockingQueue<>(1024*1024);
    private IVoucherOrderService proxy;
    static {
        asyncMiaoShaScript=new DefaultRedisScript<>();
        asyncMiaoShaScript.setLocation(new ClassPathResource("asyncMiaoSha.lua"));
        asyncMiaoShaScript.setResultType(Long.TYPE);
    }
    @PostConstruct
    public void init(){
        stringRedisTemplate.delete("orders.create");
        //为某个消息队列创建消费组
        stringRedisTemplate.opsForStream().createGroup("orders.create","ocg");
        //阻塞等待消息并处理
        miaoshaExecutor.submit(new Runnable() {
            @Override
            public void run() {
               /* while (true){
                    try {
                        VoucherOrder take = blockingQueue.take();
                        proxy.createOrder(take);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }*/
                try {
                    while (true){
                    List<MapRecord<String, Object, Object>> read = stringRedisTemplate.opsForStream().read(
                            Consumer.from("ocg","occ")
                            , StreamReadOptions.empty().count(1).block(Duration.ofSeconds(6L))
                            , StreamOffset.create("orders.create", ReadOffset.lastConsumed())
                    );
                    if (read==null||read.isEmpty())continue;
                        MapRecord<String, Object, Object> entry = read.get(0);
                        Map<Object, Object> orderMap = entry.getValue();
                        VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(orderMap, new VoucherOrder(), true);
                        //创建订单
                        proxy.createOrder(voucherOrder);
                        //ack
                        stringRedisTemplate.opsForStream().acknowledge("orders.create","ocg",entry.getId());
                    }
                } catch (Exception e) {
                    log.error("订单noAck"+e);
                    //处理异常消息 确保消息被消费
                    disposePaddingList();
                }
            }

            private void disposePaddingList()   {
                try {
                    while (true){
                        List<MapRecord<String, Object, Object>> read = stringRedisTemplate.opsForStream().read(
                                Consumer.from("oc","occ")
                                , StreamReadOptions.empty().count(1)
                                , StreamOffset.create("orders.create", ReadOffset.from("0"))
                        );
                        if (read==null||read.isEmpty())break;
                        MapRecord<String, Object, Object> entry = read.get(0);
                        Map<Object, Object> orderMap = entry.getValue();
                        VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(orderMap, new VoucherOrder(), true);
                        //创建订单
                        proxy.createOrder(voucherOrder);
                        //ack
                        stringRedisTemplate.opsForStream().acknowledge("orders.create","ocg",entry.getId());
                    }
                } catch (Exception e) {
                    log.error("异常订单noAck"+e);
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        });
    }
    //基于redis做异步秒杀
    @Override
    public Result addVoucherSeckillOrder(Long voucherId) {
        //一.基于lua脚本实现判断是否秒杀成功成功
        long orderId = overallIdCreate.generateOverAllId("miaosha" + voucherId);
        Long execute = stringRedisTemplate.execute(asyncMiaoShaScript, Collections.emptyList(), UserHolder.getUser().getId().toString(), voucherId.toString(),String.valueOf(orderId));
        if(execute==-1){
            return  Result.fail("一人只限购一单");
        }
        if(execute==-2){
            return  Result.fail("抢完了");
        }
        //...
        //二.异步去创建订单
        if(proxy==null) proxy=(IVoucherOrderService) AopContext.currentProxy();
        return Result.ok(orderId);
    }

//    @Override
   /* public Result addVoucherSeckillOrder(Long voucherId) {
        //1.根据voucherID查询voucher信息
        SeckillVoucher seckillVoucher = iSeckillVoucherService.getById(voucherId);
        //2.查看是否处于秒杀时间内
        LocalDateTime now = LocalDateTime.now();
        if (now.isBefore(seckillVoucher.getBeginTime()) || now.isAfter(seckillVoucher.getEndTime())) {
            return Result.fail("未到开放时间");
        }
        //3.查看库存是否充足
        Integer stock = seckillVoucher.getStock();
        if (stock < 1) {
            return Result.fail("抢光了!!!)");
        }
       *//* //利用java提供synchronize的实现一人一单
        //获取用户id
        Long userId = UserHolder.getUser().getId();
        synchronized (userId.toString().intern()){//为同一个用户加锁实现一人一单
            //如果直接调用 createOrder() 相当于this.createOrder() 那么@Transactional将失效 所以要先获得代理对象
            IVoucherOrderService proxy =(IVoucherOrderService) AopContext.currentProxy();
            return proxy.createOrder(voucherId, userId);
        }*//*
//        return createOrder(voucherId,userId);
        //利用redis实现分布式下的一人一单
        Long userId = UserHolder.getUser().getId();
        MyILockImpl myILock = new MyILockImpl("OpOo"+userId, stringRedisTemplate, 10);
        if(!myILock.lock()){
            return Result.fail("不要重复抢单");
        }
        try {
            //如果直接调用 createOrder() 相当于this.createOrder() 那么@Transactional将失效 所以要先获得代理对象
            IVoucherOrderService proxy =(IVoucherOrderService) AopContext.currentProxy();
            return proxy.createOrder(voucherId, userId);
        } catch (IllegalStateException e) {
            throw new RuntimeException();
        }finally {
            myILock.unlock();
        }
    }*/
    @Transactional
    public void createOrder(VoucherOrder voucherOrder) {
        long id;
            //查看是否已经抢到单
            Integer count = query().eq("user_id", voucherOrder.getUserId()).eq("voucher_id",voucherOrder.getVoucherId()).count();
            if(count>0){
                log.error("重复下单");
                return;
            }
            //3.1扣减库存
            boolean ok = iSeckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();
            if(!ok){
                log.error("抢光了!!!)");
                return;
            }
    /*        //4.创建订单
            VoucherOrder voucherOrder = new VoucherOrder();
           //4.1生成全局唯一订单id
            id = overallIdCreate.generateOverAllId("voucher:order");
            voucherOrder.setId(id);
            voucherOrder.setVoucherId(voucherId);
            voucherOrder.setUserId(userId);*/
            //5.新增订单
            save(voucherOrder);
        //返回订单id
    }
}
